import logging
import pandas as pd
import numpy as np
from typing import List, Dict, Any, Optional, Tuple, Callable, Union, Type
import time
import os
import pickle
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import seaborn as sns

# 尝试导入金融数据API
try:
    import akshare as ak
    AKSHARE_AVAILABLE = True
except ImportError:
    AKSHARE_AVAILABLE = False
    logging.warning("AKShare未安装，部分基本面因子可能无法使用")

try:
    import tushare as ts
    TUSHARE_AVAILABLE = True
except ImportError:
    TUSHARE_AVAILABLE = False
    logging.warning("Tushare未安装，部分基本面因子可能无法使用")


class FinancialDataProvider:
    """金融数据提供者基类，提供财务数据获取和缓存功能"""
    
    def __init__(self, cache_dir: str = "./data_cache/fundamental", cache_ttl: int = 86400 * 7):
        """
        初始化金融数据提供者
        
        参数:
            cache_dir: 缓存目录
            cache_ttl: 缓存有效期（秒）
        """
        self.cache_dir = cache_dir
        self.cache_ttl = cache_ttl
        
        # 创建缓存目录
        os.makedirs(cache_dir, exist_ok=True)
        
    def _get_cache_path(self, data_type: str, params: Dict[str, Any]) -> str:
        """
        获取缓存文件路径
        
        参数:
            data_type: 数据类型
            params: 参数字典
            
        返回:
            str: 缓存文件路径
        """
        # 将参数转换为排序后的字符串
        param_str = "_".join(f"{k}={v}" for k, v in sorted(params.items()))
        return os.path.join(self.cache_dir, f"{data_type}_{param_str}.pkl")
        
    def _load_from_cache(self, data_type: str, params: Dict[str, Any]) -> Optional[pd.DataFrame]:
        """
        从缓存加载数据
        
        参数:
            data_type: 数据类型
            params: 参数字典
            
        返回:
            Optional[pd.DataFrame]: 数据，如果缓存不存在或过期则返回None
        """
        cache_path = self._get_cache_path(data_type, params)
        
        if not os.path.exists(cache_path):
            return None
            
        # 检查缓存是否过期
        if self.cache_ttl > 0:
            mtime = os.path.getmtime(cache_path)
            if time.time() - mtime > self.cache_ttl:
                return None
                
        try:
            with open(cache_path, 'rb') as f:
                return pickle.load(f)
        except Exception as e:
            logging.warning(f"加载缓存失败: {str(e)}")
            return None
            
    def _save_to_cache(self, data: pd.DataFrame, data_type: str, params: Dict[str, Any]) -> None:
        """
        保存数据到缓存
        
        参数:
            data: 数据
            data_type: 数据类型
            params: 参数字典
        """
        if data is None or data.empty:
            logging.warning(f"数据为空，不缓存 {data_type}")
            return
            
        cache_path = self._get_cache_path(data_type, params)
        
        try:
            with open(cache_path, 'wb') as f:
                pickle.dump(data, f)
        except Exception as e:
            logging.warning(f"保存缓存失败: {str(e)}")
            
    def get_financial_data(self, data_type: str, **params) -> Optional[pd.DataFrame]:
        """
        获取金融数据，优先从缓存加载
        
        参数:
            data_type: 数据类型
            **params: 参数
            
        返回:
            Optional[pd.DataFrame]: 数据
        """
        # 尝试从缓存加载
        data = self._load_from_cache(data_type, params)
        if data is not None:
            return data
            
        # 根据数据类型获取数据
        method_name = f"_get_{data_type}"
        if hasattr(self, method_name):
            try:
                data = getattr(self, method_name)(**params)
                
                # 缓存数据
                if data is not None and not data.empty:
                    self._save_to_cache(data, data_type, params)
                    
                return data
            except Exception as e:
                logging.error(f"获取{data_type}数据失败: {str(e)}")
                return None
        else:
            logging.error(f"未知的数据类型: {data_type}")
            return None


class AKShareFinancialDataProvider(FinancialDataProvider):
    """使用AKShare获取财务数据"""
    
    def __init__(self, cache_dir: str = "./data_cache/fundamental", cache_ttl: int = 86400 * 7):
        """初始化AKShare数据提供者"""
        super().__init__(cache_dir, cache_ttl)
        
        if not AKSHARE_AVAILABLE:
            raise ImportError("AKShare未安装，请使用pip install akshare安装")
            
    def _get_income_statement(self, symbol: str, period: str = "annual") -> pd.DataFrame:
        """
        获取利润表数据
        
        参数:
            symbol: 股票代码
            period: 报告期，可选值：'annual', 'quarterly'
            
        返回:
            pd.DataFrame: 利润表数据
        """
        try:
            # 添加市场后缀
            if symbol.startswith('6'):
                full_symbol = f"{symbol}.SH"
            else:
                full_symbol = f"{symbol}.SZ"
                
            # 获取利润表
            if period == "annual":
                df = ak.stock_financial_report_sina(symbol=full_symbol, symbol_type="利润表")
            else:
                df = ak.stock_financial_report_sina(symbol=full_symbol, symbol_type="季度利润表")
                
            return df
        except Exception as e:
            logging.error(f"获取利润表数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_balance_sheet(self, symbol: str, period: str = "annual") -> pd.DataFrame:
        """
        获取资产负债表数据
        
        参数:
            symbol: 股票代码
            period: 报告期，可选值：'annual', 'quarterly'
            
        返回:
            pd.DataFrame: 资产负债表数据
        """
        try:
            # 添加市场后缀
            if symbol.startswith('6'):
                full_symbol = f"{symbol}.SH"
            else:
                full_symbol = f"{symbol}.SZ"
                
            # 获取资产负债表
            if period == "annual":
                df = ak.stock_financial_report_sina(symbol=full_symbol, symbol_type="资产负债表")
            else:
                df = ak.stock_financial_report_sina(symbol=full_symbol, symbol_type="季度资产负债表")
                
            return df
        except Exception as e:
            logging.error(f"获取资产负债表数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_cash_flow(self, symbol: str, period: str = "annual") -> pd.DataFrame:
        """
        获取现金流量表数据
        
        参数:
            symbol: 股票代码
            period: 报告期，可选值：'annual', 'quarterly'
            
        返回:
            pd.DataFrame: 现金流量表数据
        """
        try:
            # 添加市场后缀
            if symbol.startswith('6'):
                full_symbol = f"{symbol}.SH"
            else:
                full_symbol = f"{symbol}.SZ"
                
            # 获取现金流量表
            if period == "annual":
                df = ak.stock_financial_report_sina(symbol=full_symbol, symbol_type="现金流量表")
            else:
                df = ak.stock_financial_report_sina(symbol=full_symbol, symbol_type="季度现金流量表")
                
            return df
        except Exception as e:
            logging.error(f"获取现金流量表数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_financial_indicator(self, symbol: str) -> pd.DataFrame:
        """
        获取财务指标数据
        
        参数:
            symbol: 股票代码
            
        返回:
            pd.DataFrame: 财务指标数据
        """
        try:
            # 获取财务指标
            df = ak.stock_financial_analysis_indicator(symbol=symbol)
            return df
        except Exception as e:
            logging.error(f"获取财务指标数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_valuation(self, symbol: str) -> pd.DataFrame:
        """
        获取估值数据
        
        参数:
            symbol: 股票代码
            
        返回:
            pd.DataFrame: 估值数据
        """
        try:
            # 获取PE、PB等估值数据
            if symbol.startswith('6'):
                exchange = 'sh'
            else:
                exchange = 'sz'
                
            df_sh = ak.stock_zh_a_spot_em()
            df = df_sh[df_sh['代码'] == f"{exchange}{symbol}"]
            
            if df.empty:
                return pd.DataFrame()
                
            # 重命名列
            df = df.rename(columns={
                '代码': 'symbol',
                '名称': 'name',
                '最新价': 'price',
                '涨跌幅': 'change_pct',
                '市盈率-动态': 'pe_ttm',
                '市净率': 'pb',
                '总市值': 'market_cap',
                '流通市值': 'float_market_cap'
            })
            
            return df
        except Exception as e:
            logging.error(f"获取估值数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_dividend(self, symbol: str) -> pd.DataFrame:
        """
        获取分红数据
        
        参数:
            symbol: 股票代码
            
        返回:
            pd.DataFrame: 分红数据
        """
        try:
            # 获取分红数据
            df = ak.stock_history_dividend_detail(symbol=symbol)
            return df
        except Exception as e:
            logging.error(f"获取分红数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_industry(self, symbol: str = None) -> pd.DataFrame:
        """
        获取行业数据
        
        参数:
            symbol: 股票代码
            
        返回:
            pd.DataFrame: 行业数据
        """
        try:
            # 获取行业数据
            df = ak.stock_sector_spot()
            
            if symbol is not None:
                # 过滤指定股票
                if symbol.startswith('6'):
                    full_symbol = f"sh{symbol}"
                else:
                    full_symbol = f"sz{symbol}"
                    
                df = df[df['代码'] == full_symbol]
                
            return df
        except Exception as e:
            logging.error(f"获取行业数据失败: {str(e)}")
            return pd.DataFrame()


class TushareFinancialDataProvider(FinancialDataProvider):
    """使用Tushare获取财务数据"""
    
    def __init__(self, token: str, cache_dir: str = "./data_cache/fundamental", cache_ttl: int = 86400 * 7):
        """
        初始化Tushare数据提供者
        
        参数:
            token: Tushare API令牌
            cache_dir: 缓存目录
            cache_ttl: 缓存有效期（秒）
        """
        super().__init__(cache_dir, cache_ttl)
        
        if not TUSHARE_AVAILABLE:
            raise ImportError("Tushare未安装，请使用pip install tushare安装")
            
        # 设置Tushare TOKEN
        ts.set_token(token)
        self.pro = ts.pro_api()
        
    def _get_income_statement(self, symbol: str, period: str = "annual", start_date: str = None, end_date: str = None) -> pd.DataFrame:
        """
        获取利润表数据
        
        参数:
            symbol: 股票代码
            period: 报告期，可选值：'annual', 'quarterly'
            start_date: 开始日期
            end_date: 结束日期
            
        返回:
            pd.DataFrame: 利润表数据
        """
        try:
            # 处理日期
            if end_date is None:
                end_date = datetime.now().strftime('%Y%m%d')
            if start_date is None:
                start_date = (datetime.now() - timedelta(days=365*5)).strftime('%Y%m%d')
                
            # 获取利润表
            if period == "annual":
                df = self.pro.income(ts_code=symbol, start_date=start_date, end_date=end_date, period='00')
            else:
                df = self.pro.income(ts_code=symbol, start_date=start_date, end_date=end_date)
                
            return df
        except Exception as e:
            logging.error(f"获取利润表数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_balance_sheet(self, symbol: str, period: str = "annual", start_date: str = None, end_date: str = None) -> pd.DataFrame:
        """
        获取资产负债表数据
        
        参数:
            symbol: 股票代码
            period: 报告期，可选值：'annual', 'quarterly'
            start_date: 开始日期
            end_date: 结束日期
            
        返回:
            pd.DataFrame: 资产负债表数据
        """
        try:
            # 处理日期
            if end_date is None:
                end_date = datetime.now().strftime('%Y%m%d')
            if start_date is None:
                start_date = (datetime.now() - timedelta(days=365*5)).strftime('%Y%m%d')
                
            # 获取资产负债表
            if period == "annual":
                df = self.pro.balancesheet(ts_code=symbol, start_date=start_date, end_date=end_date, period='00')
            else:
                df = self.pro.balancesheet(ts_code=symbol, start_date=start_date, end_date=end_date)
                
            return df
        except Exception as e:
            logging.error(f"获取资产负债表数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_cash_flow(self, symbol: str, period: str = "annual", start_date: str = None, end_date: str = None) -> pd.DataFrame:
        """
        获取现金流量表数据
        
        参数:
            symbol: 股票代码
            period: 报告期，可选值：'annual', 'quarterly'
            start_date: 开始日期
            end_date: 结束日期
            
        返回:
            pd.DataFrame: 现金流量表数据
        """
        try:
            # 处理日期
            if end_date is None:
                end_date = datetime.now().strftime('%Y%m%d')
            if start_date is None:
                start_date = (datetime.now() - timedelta(days=365*5)).strftime('%Y%m%d')
                
            # 获取现金流量表
            if period == "annual":
                df = self.pro.cashflow(ts_code=symbol, start_date=start_date, end_date=end_date, period='00')
            else:
                df = self.pro.cashflow(ts_code=symbol, start_date=start_date, end_date=end_date)
                
            return df
        except Exception as e:
            logging.error(f"获取现金流量表数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_financial_indicator(self, symbol: str, start_date: str = None, end_date: str = None) -> pd.DataFrame:
        """
        获取财务指标数据
        
        参数:
            symbol: 股票代码
            start_date: 开始日期
            end_date: 结束日期
            
        返回:
            pd.DataFrame: 财务指标数据
        """
        try:
            # 处理日期
            if end_date is None:
                end_date = datetime.now().strftime('%Y%m%d')
            if start_date is None:
                start_date = (datetime.now() - timedelta(days=365*5)).strftime('%Y%m%d')
                
            # 获取财务指标
            df = self.pro.fina_indicator(ts_code=symbol, start_date=start_date, end_date=end_date)
            return df
        except Exception as e:
            logging.error(f"获取财务指标数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_valuation(self, symbol: str, start_date: str = None, end_date: str = None) -> pd.DataFrame:
        """
        获取估值数据
        
        参数:
            symbol: 股票代码
            start_date: 开始日期
            end_date: 结束日期
            
        返回:
            pd.DataFrame: 估值数据
        """
        try:
            # 处理日期
            if end_date is None:
                end_date = datetime.now().strftime('%Y%m%d')
            if start_date is None:
                start_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d')
                
            # 获取每日指标
            df = self.pro.daily_basic(ts_code=symbol, start_date=start_date, end_date=end_date,
                                    fields='ts_code,trade_date,close,pe,pe_ttm,pb,ps,ps_ttm,dv_ratio,dv_ttm,total_mv,circ_mv')
            return df
        except Exception as e:
            logging.error(f"获取估值数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_dividend(self, symbol: str) -> pd.DataFrame:
        """
        获取分红数据
        
        参数:
            symbol: 股票代码
            
        返回:
            pd.DataFrame: 分红数据
        """
        try:
            # 获取分红数据
            df = self.pro.dividend(ts_code=symbol)
            return df
        except Exception as e:
            logging.error(f"获取分红数据失败: {str(e)}")
            return pd.DataFrame()
            
    def _get_industry(self, symbol: str = None) -> pd.DataFrame:
        """
        获取行业数据
        
        参数:
            symbol: 股票代码
            
        返回:
            pd.DataFrame: 行业数据
        """
        try:
            if symbol is not None:
                # 获取单个股票行业
                df = self.pro.stock_basic(ts_code=symbol, fields='ts_code,symbol,name,area,industry,list_date')
            else:
                # 获取所有股票行业
                df = self.pro.stock_basic(fields='ts_code,symbol,name,area,industry,list_date')
                
            return df
        except Exception as e:
            logging.error(f"获取行业数据失败: {str(e)}")
            return pd.DataFrame()


class FundamentalFactorCalculator:
    """基本面因子计算器"""
    
    def __init__(self, data_provider: FinancialDataProvider):
        """
        初始化基本面因子计算器
        
        参数:
            data_provider: 金融数据提供者
        """
        self.data_provider = data_provider
        
    def calculate_valuation_factors(self, symbol: str, price_data: pd.DataFrame = None, 
                                   align_price: bool = True) -> Dict[str, pd.Series]:
        """
        计算估值类因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            align_price: 是否将财务数据对齐到价格日期
            
        返回:
            Dict[str, pd.Series]: 因子名称到因子值的映射
        """
        factors = {}
        
        # 获取估值数据
        valuation = self.data_provider.get_financial_data("valuation", symbol=symbol)
        
        if valuation is None or valuation.empty:
            logging.warning(f"无法获取 {symbol} 的估值数据")
            return factors
            
        # 如果需要对齐价格数据
        if align_price and price_data is not None:
            # 确保估值数据有日期索引
            if not isinstance(valuation.index, pd.DatetimeIndex):
                if 'trade_date' in valuation.columns:
                    valuation['trade_date'] = pd.to_datetime(valuation['trade_date'])
                    valuation.set_index('trade_date', inplace=True)
                elif 'ann_date' in valuation.columns:
                    valuation['ann_date'] = pd.to_datetime(valuation['ann_date'])
                    valuation.set_index('ann_date', inplace=True)
                    
            # 对齐到价格数据索引
            aligned_data = pd.DataFrame(index=price_data.index)
            
            # 前向填充估值数据
            for col in ['pe_ttm', 'pb', 'ps_ttm', 'dv_ttm']:
                if col in valuation.columns:
                    aligned_data[col] = None
                    for idx in aligned_data.index:
                        valid_data = valuation.loc[valuation.index <= idx, col]
                        if not valid_data.empty:
                            aligned_data.loc[idx, col] = valid_data.iloc[-1]
                            
            # 创建因子
            for col in ['pe_ttm', 'pb', 'ps_ttm', 'dv_ttm']:
                if col in aligned_data.columns:
                    factors[col] = aligned_data[col]
                    
            # 计算EP(盈利收益率)因子 = 1/PE
            if 'pe_ttm' in aligned_data.columns:
                factors['ep'] = 1 / aligned_data['pe_ttm']
        else:
            # 不需要对齐，直接使用估值数据
            for col in ['pe_ttm', 'pb', 'ps_ttm', 'dv_ttm']:
                if col in valuation.columns:
                    factors[col] = valuation[col]
                    
            # 计算EP(盈利收益率)因子 = 1/PE
            if 'pe_ttm' in valuation.columns:
                factors['ep'] = 1 / valuation['pe_ttm']
                
        return factors
        
    def calculate_financial_factors(self, symbol: str, price_data: pd.DataFrame = None, 
                                   align_price: bool = True) -> Dict[str, pd.Series]:
        """
        计算财务类因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            align_price: 是否将财务数据对齐到价格日期
            
        返回:
            Dict[str, pd.Series]: 因子名称到因子值的映射
        """
        factors = {}
        
        # 获取财务指标数据
        financial_indicator = self.data_provider.get_financial_data("financial_indicator", symbol=symbol)
        
        if financial_indicator is None or financial_indicator.empty:
            logging.warning(f"无法获取 {symbol} 的财务指标数据")
            return factors
            
        # 如果需要对齐价格数据
        if align_price and price_data is not None:
            # 确保财务数据有日期索引
            if not isinstance(financial_indicator.index, pd.DatetimeIndex):
                if 'end_date' in financial_indicator.columns:
                    financial_indicator['end_date'] = pd.to_datetime(financial_indicator['end_date'])
                    financial_indicator.set_index('end_date', inplace=True)
                elif 'ann_date' in financial_indicator.columns:
                    financial_indicator['ann_date'] = pd.to_datetime(financial_indicator['ann_date'])
                    financial_indicator.set_index('ann_date', inplace=True)
                    
            # 对齐到价格数据索引
            aligned_data = pd.DataFrame(index=price_data.index)
            
            # 前向填充财务数据
            financial_columns = [
                'roe', 'roa', 'grossprofit_margin', 'netprofit_margin',
                'debt_to_assets', 'current_ratio', 'quick_ratio',
                'fcff', 'asset_turn', 'op_income_yoy', 'tr_yoy', 'or_yoy', 'netprofit_yoy'
            ]
            
            for col in financial_columns:
                if col in financial_indicator.columns:
                    aligned_data[col] = None
                    for idx in aligned_data.index:
                        valid_data = financial_indicator.loc[financial_indicator.index <= idx, col]
                        if not valid_data.empty:
                            aligned_data.loc[idx, col] = valid_data.iloc[-1]
                            
            # 创建因子
            for col in financial_columns:
                if col in aligned_data.columns:
                    factors[col] = aligned_data[col]
        else:
            # 不需要对齐，直接使用财务数据
            financial_columns = [
                'roe', 'roa', 'grossprofit_margin', 'netprofit_margin',
                'debt_to_assets', 'current_ratio', 'quick_ratio',
                'fcff', 'asset_turn', 'op_income_yoy', 'tr_yoy', 'or_yoy', 'netprofit_yoy'
            ]
            
            for col in financial_columns:
                if col in financial_indicator.columns:
                    factors[col] = financial_indicator[col]
                    
        return factors
        
    def calculate_growth_factors(self, symbol: str, price_data: pd.DataFrame = None, 
                              align_price: bool = True) -> Dict[str, pd.Series]:
        """
        计算成长类因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            align_price: 是否将财务数据对齐到价格日期
            
        返回:
            Dict[str, pd.Series]: 因子名称到因子值的映射
        """
        factors = {}
        
        # 获取财务指标数据
        financial_indicator = self.data_provider.get_financial_data("financial_indicator", symbol=symbol)
        
        if financial_indicator is None or financial_indicator.empty:
            logging.warning(f"无法获取 {symbol} 的财务指标数据")
            return factors
            
        # 如果需要对齐价格数据
        if align_price and price_data is not None:
            # 确保财务数据有日期索引
            if not isinstance(financial_indicator.index, pd.DatetimeIndex):
                if 'end_date' in financial_indicator.columns:
                    financial_indicator['end_date'] = pd.to_datetime(financial_indicator['end_date'])
                    financial_indicator.set_index('end_date', inplace=True)
                elif 'ann_date' in financial_indicator.columns:
                    financial_indicator['ann_date'] = pd.to_datetime(financial_indicator['ann_date'])
                    financial_indicator.set_index('ann_date', inplace=True)
                    
            # 对齐到价格数据索引
            aligned_data = pd.DataFrame(index=price_data.index)
            
            # 前向填充财务数据
            growth_columns = [
                'netprofit_yoy', 'tr_yoy', 'or_yoy', 'op_yoy',
                'netprofit_growth', 'tr_growth', 'or_growth', 'op_growth'
            ]
            
            for col in growth_columns:
                if col in financial_indicator.columns:
                    aligned_data[col] = None
                    for idx in aligned_data.index:
                        valid_data = financial_indicator.loc[financial_indicator.index <= idx, col]
                        if not valid_data.empty:
                            aligned_data.loc[idx, col] = valid_data.iloc[-1]
                            
            # 创建因子
            for col in growth_columns:
                if col in aligned_data.columns:
                    factors[col] = aligned_data[col]
        else:
            # 不需要对齐，直接使用财务数据
            growth_columns = [
                'netprofit_yoy', 'tr_yoy', 'or_yoy', 'op_yoy',
                'netprofit_growth', 'tr_growth', 'or_growth', 'op_growth'
            ]
            
            for col in growth_columns:
                if col in financial_indicator.columns:
                    factors[col] = financial_indicator[col]
                    
        return factors
        
    def calculate_quality_factors(self, symbol: str, price_data: pd.DataFrame = None, 
                               align_price: bool = True) -> Dict[str, pd.Series]:
        """
        计算质量类因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            align_price: 是否将财务数据对齐到价格日期
            
        返回:
            Dict[str, pd.Series]: 因子名称到因子值的映射
        """
        factors = {}
        
        # 获取财务指标数据
        financial_indicator = self.data_provider.get_financial_data("financial_indicator", symbol=symbol)
        income_statement = self.data_provider.get_financial_data("income_statement", symbol=symbol)
        cash_flow = self.data_provider.get_financial_data("cash_flow", symbol=symbol)
        
        if (financial_indicator is None or financial_indicator.empty) and \
           (income_statement is None or income_statement.empty) and \
           (cash_flow is None or cash_flow.empty):
            logging.warning(f"无法获取 {symbol} 的财务数据")
            return factors
            
        # 如果需要对齐价格数据
        if align_price and price_data is not None:
            # 处理财务指标数据
            if financial_indicator is not None and not financial_indicator.empty:
                # 确保财务数据有日期索引
                if not isinstance(financial_indicator.index, pd.DatetimeIndex):
                    if 'end_date' in financial_indicator.columns:
                        financial_indicator['end_date'] = pd.to_datetime(financial_indicator['end_date'])
                        financial_indicator.set_index('end_date', inplace=True)
                    elif 'ann_date' in financial_indicator.columns:
                        financial_indicator['ann_date'] = pd.to_datetime(financial_indicator['ann_date'])
                        financial_indicator.set_index('ann_date', inplace=True)
                        
                # 对齐到价格数据索引
                aligned_data = pd.DataFrame(index=price_data.index)
                
                # 前向填充财务数据
                quality_columns = [
                    'roe_yearly', 'roa_yearly', 'asset_quality', 'debt_to_assets',
                    'op_of_gr', 'roe_waa', 'roe_dt'
                ]
                
                for col in quality_columns:
                    if col in financial_indicator.columns:
                        aligned_data[col] = None
                        for idx in aligned_data.index:
                            valid_data = financial_indicator.loc[financial_indicator.index <= idx, col]
                            if not valid_data.empty:
                                aligned_data.loc[idx, col] = valid_data.iloc[-1]
                                
                # 创建因子
                for col in quality_columns:
                    if col in aligned_data.columns:
                        factors[col] = aligned_data[col]
                        
            # 处理现金流量表数据
            if cash_flow is not None and not cash_flow.empty and income_statement is not None and not income_statement.empty:
                # 确保现金流量表数据有日期索引
                if not isinstance(cash_flow.index, pd.DatetimeIndex):
                    if 'end_date' in cash_flow.columns:
                        cash_flow['end_date'] = pd.to_datetime(cash_flow['end_date'])
                        cash_flow.set_index('end_date', inplace=True)
                    elif 'ann_date' in cash_flow.columns:
                        cash_flow['ann_date'] = pd.to_datetime(cash_flow['ann_date'])
                        cash_flow.set_index('ann_date', inplace=True)
                        
                # 确保利润表数据有日期索引
                if not isinstance(income_statement.index, pd.DatetimeIndex):
                    if 'end_date' in income_statement.columns:
                        income_statement['end_date'] = pd.to_datetime(income_statement['end_date'])
                        income_statement.set_index('end_date', inplace=True)
                    elif 'ann_date' in income_statement.columns:
                        income_statement['ann_date'] = pd.to_datetime(income_statement['ann_date'])
                        income_statement.set_index('ann_date', inplace=True)
                        
                # 计算现金流质量因子
                if 'n_cashflow_act' in cash_flow.columns and 'n_income' in income_statement.columns:
                    # 合并数据
                    cf_quality = pd.merge(
                        cash_flow[['n_cashflow_act']],
                        income_statement[['n_income']],
                        left_index=True,
                        right_index=True,
                        how='inner'
                    )
                    
                    # 计算现金流质量
                    cf_quality['cash_flow_quality'] = cf_quality['n_cashflow_act'] / cf_quality['n_income']
                    
                    # 对齐到价格数据索引
                    aligned_cf = pd.DataFrame(index=price_data.index)
                    aligned_cf['cash_flow_quality'] = None
                    
                    for idx in aligned_cf.index:
                        valid_data = cf_quality.loc[cf_quality.index <= idx, 'cash_flow_quality']
                        if not valid_data.empty:
                            aligned_cf.loc[idx, 'cash_flow_quality'] = valid_data.iloc[-1]
                            
                    factors['cash_flow_quality'] = aligned_cf['cash_flow_quality']
        else:
            # 不需要对齐，直接使用财务数据
            if financial_indicator is not None and not financial_indicator.empty:
                quality_columns = [
                    'roe_yearly', 'roa_yearly', 'asset_quality', 'debt_to_assets',
                    'op_of_gr', 'roe_waa', 'roe_dt'
                ]
                
                for col in quality_columns:
                    if col in financial_indicator.columns:
                        factors[col] = financial_indicator[col]
                        
            # 计算现金流质量因子
            if cash_flow is not None and not cash_flow.empty and income_statement is not None and not income_statement.empty:
                if 'n_cashflow_act' in cash_flow.columns and 'n_income' in income_statement.columns:
                    # 合并数据
                    cf_quality = pd.merge(
                        cash_flow[['n_cashflow_act']],
                        income_statement[['n_income']],
                        left_index=True,
                        right_index=True,
                        how='inner'
                    )
                    
                    # 计算现金流质量
                    cf_quality['cash_flow_quality'] = cf_quality['n_cashflow_act'] / cf_quality['n_income']
                    factors['cash_flow_quality'] = cf_quality['cash_flow_quality']
                    
        return factors
        
    def calculate_dividend_factors(self, symbol: str, price_data: pd.DataFrame = None, 
                                align_price: bool = True) -> Dict[str, pd.Series]:
        """
        计算股息类因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            align_price: 是否将财务数据对齐到价格日期
            
        返回:
            Dict[str, pd.Series]: 因子名称到因子值的映射
        """
        factors = {}
        
        # 获取分红数据
        dividend = self.data_provider.get_financial_data("dividend", symbol=symbol)
        valuation = self.data_provider.get_financial_data("valuation", symbol=symbol)
        
        if (dividend is None or dividend.empty) and (valuation is None or valuation.empty):
            logging.warning(f"无法获取 {symbol} 的分红数据")
            return factors
            
        # 从估值数据中提取股息率
        if valuation is not None and not valuation.empty:
            if 'dv_ttm' in valuation.columns:
                if align_price and price_data is not None:
                    # 确保估值数据有日期索引
                    if not isinstance(valuation.index, pd.DatetimeIndex):
                        if 'trade_date' in valuation.columns:
                            valuation['trade_date'] = pd.to_datetime(valuation['trade_date'])
                            valuation.set_index('trade_date', inplace=True)
                        elif 'ann_date' in valuation.columns:
                            valuation['ann_date'] = pd.to_datetime(valuation['ann_date'])
                            valuation.set_index('ann_date', inplace=True)
                            
                    # 对齐到价格数据索引
                    aligned_data = pd.DataFrame(index=price_data.index)
                    aligned_data['dv_ttm'] = None
                    
                    for idx in aligned_data.index:
                        valid_data = valuation.loc[valuation.index <= idx, 'dv_ttm']
                        if not valid_data.empty:
                            aligned_data.loc[idx, 'dv_ttm'] = valid_data.iloc[-1]
                            
                    factors['dividend_yield'] = aligned_data['dv_ttm']
                else:
                    factors['dividend_yield'] = valuation['dv_ttm']
                    
        # 处理分红历史数据
        if dividend is not None and not dividend.empty:
            # 确保分红数据有日期索引
            if not isinstance(dividend.index, pd.DatetimeIndex):
                if 'div_proc' in dividend.columns:
                    dividend['div_proc'] = pd.to_datetime(dividend['div_proc'])
                    dividend.set_index('div_proc', inplace=True)
                elif 'announcement_date' in dividend.columns:
                    dividend['announcement_date'] = pd.to_datetime(dividend['announcement_date'])
                    dividend.set_index('announcement_date', inplace=True)
                    
            # 计算过去N年的分红次数
            if align_price and price_data is not None:
                # 对齐到价格数据索引
                aligned_data = pd.DataFrame(index=price_data.index)
                aligned_data['dividend_count_3y'] = 0
                aligned_data['dividend_count_5y'] = 0
                aligned_data['dividend_consistency'] = 0
                
                for idx in aligned_data.index:
                    # 计算3年内分红次数
                    start_date_3y = idx - pd.DateOffset(years=3)
                    dividend_3y = dividend.loc[(dividend.index <= idx) & (dividend.index > start_date_3y)]
                    aligned_data.loc[idx, 'dividend_count_3y'] = len(dividend_3y)
                    
                    # 计算5年内分红次数
                    start_date_5y = idx - pd.DateOffset(years=5)
                    dividend_5y = dividend.loc[(dividend.index <= idx) & (dividend.index > start_date_5y)]
                    aligned_data.loc[idx, 'dividend_count_5y'] = len(dividend_5y)
                    
                    # 计算分红一致性（过去5年每年是否都有分红）
                    if len(dividend_5y) > 0:
                        years = [d.year for d in dividend_5y.index]
                        unique_years = len(set(years))
                        aligned_data.loc[idx, 'dividend_consistency'] = unique_years / 5
                        
                factors['dividend_count_3y'] = aligned_data['dividend_count_3y']
                factors['dividend_count_5y'] = aligned_data['dividend_count_5y']
                factors['dividend_consistency'] = aligned_data['dividend_consistency']
                
        return factors
        
    def calculate_all_fundamental_factors(self, symbol: str, price_data: pd.DataFrame = None, 
                                        align_price: bool = True) -> Dict[str, pd.Series]:
        """
        计算所有基本面因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            align_price: 是否将财务数据对齐到价格日期
            
        返回:
            Dict[str, pd.Series]: 因子名称到因子值的映射
        """
        all_factors = {}
        
        # 计算各类因子
        valuation_factors = self.calculate_valuation_factors(symbol, price_data, align_price)
        financial_factors = self.calculate_financial_factors(symbol, price_data, align_price)
        growth_factors = self.calculate_growth_factors(symbol, price_data, align_price)
        quality_factors = self.calculate_quality_factors(symbol, price_data, align_price)
        dividend_factors = self.calculate_dividend_factors(symbol, price_data, align_price)
        
        # 合并所有因子
        all_factors.update(valuation_factors)
        all_factors.update(financial_factors)
        all_factors.update(growth_factors)
        all_factors.update(quality_factors)
        all_factors.update(dividend_factors)
        
        return all_factors


class FundamentalFactors:
    """基本面因子库，用于创建和管理基本面因子"""
    
    def __init__(self, data_provider: str = "akshare", **kwargs):
        """
        初始化基本面因子库
        
        参数:
            data_provider: 数据提供者，可选值：'akshare', 'tushare'
            **kwargs: 数据提供者参数
        """
        if data_provider == "akshare":
            if not AKSHARE_AVAILABLE:
                raise ImportError("AKShare未安装，请使用pip install akshare安装")
            self.provider = AKShareFinancialDataProvider(**kwargs)
        elif data_provider == "tushare":
            if not TUSHARE_AVAILABLE:
                raise ImportError("Tushare未安装，请使用pip install tushare安装")
            if 'token' not in kwargs:
                raise ValueError("使用Tushare需要提供token参数")
            self.provider = TushareFinancialDataProvider(**kwargs)
        else:
            raise ValueError(f"未知的数据提供者: {data_provider}")
            
        self.factor_calculator = FundamentalFactorCalculator(self.provider)
        
    def create_pe_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "pe_ttm",
                       description: str = "市盈率(TTM)", category: str = "估值") -> pd.Series:
        """
        创建市盈率因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_valuation_factors(symbol, price_data)
        if 'pe_ttm' in factors:
            return factors['pe_ttm']
        return None
        
    def create_pb_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "pb",
                       description: str = "市净率", category: str = "估值") -> pd.Series:
        """
        创建市净率因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_valuation_factors(symbol, price_data)
        if 'pb' in factors:
            return factors['pb']
        return None
        
    def create_ep_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "ep",
                       description: str = "盈利收益率", category: str = "估值") -> pd.Series:
        """
        创建盈利收益率因子(1/PE)
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_valuation_factors(symbol, price_data)
        if 'ep' in factors:
            return factors['ep']
        return None
        
    def create_dividend_yield_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "dividend_yield",
                                   description: str = "股息率", category: str = "股息") -> pd.Series:
        """
        创建股息率因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_dividend_factors(symbol, price_data)
        if 'dividend_yield' in factors:
            return factors['dividend_yield']
        return None
        
    def create_roe_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "roe",
                        description: str = "净资产收益率", category: str = "质量") -> pd.Series:
        """
        创建ROE因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_financial_factors(symbol, price_data)
        if 'roe' in factors:
            return factors['roe']
        return None
        
    def create_roa_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "roa",
                        description: str = "总资产收益率", category: str = "质量") -> pd.Series:
        """
        创建ROA因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_financial_factors(symbol, price_data)
        if 'roa' in factors:
            return factors['roa']
        return None
        
    def create_netprofit_growth_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "netprofit_growth",
                                    description: str = "净利润增长率", category: str = "成长") -> pd.Series:
        """
        创建净利润增长率因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_growth_factors(symbol, price_data)
        if 'netprofit_growth' in factors:
            return factors['netprofit_growth']
        return None
        
    def create_cash_flow_quality_factor(self, symbol: str, price_data: pd.DataFrame, name: str = "cash_flow_quality",
                                     description: str = "现金流质量", category: str = "质量") -> pd.Series:
        """
        创建现金流质量因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            name: 因子名称
            description: 因子描述
            category: 因子类别
            
        返回:
            pd.Series: 因子数据
        """
        factors = self.factor_calculator.calculate_quality_factors(symbol, price_data)
        if 'cash_flow_quality' in factors:
            return factors['cash_flow_quality']
        return None
        
    def create_all_fundamental_factors(self, symbol: str, price_data: pd.DataFrame) -> Dict[str, pd.Series]:
        """
        创建所有基本面因子
        
        参数:
            symbol: 股票代码
            price_data: 价格数据
            
        返回:
            Dict[str, pd.Series]: 因子名称到因子值的映射
        """
        return self.factor_calculator.calculate_all_fundamental_factors(symbol, price_data)